{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Linear Regression using Pyspark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "#create sparksession object\n",
    "from pyspark.sql import SparkSession\n",
    "spark=SparkSession.builder.appName('lin_reg').getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "#import Linear Regression from spark's MLlib\n",
    "from pyspark.ml.regression import LinearRegression"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "#Load the dataset\n",
    "df=spark.read.csv('Linear_regression_dataset.csv',inferSchema=True,header=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(1232, 6)\n"
     ]
    }
   ],
   "source": [
    "#validate the size of data\n",
    "print((df.count(), len(df.columns)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- var_1: integer (nullable = true)\n",
      " |-- var_2: integer (nullable = true)\n",
      " |-- var_3: integer (nullable = true)\n",
      " |-- var_4: double (nullable = true)\n",
      " |-- var_5: double (nullable = true)\n",
      " |-- output: double (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#explore the data\n",
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-----------------+-----------------+------------------+--------------------+--------------------+-------------------+\n",
      "|summary|var_1            |var_2            |var_3             |var_4               |var_5               |output             |\n",
      "+-------+-----------------+-----------------+------------------+--------------------+--------------------+-------------------+\n",
      "|count  |1232             |1232             |1232              |1232                |1232                |1232               |\n",
      "|mean   |715.0819805194806|715.0819805194806|80.90422077922078 |0.3263311688311693  |0.25927272727272715 |0.39734172077922014|\n",
      "|stddev |91.5342940441652 |93.07993263118064|11.458139049993724|0.015012772334166148|0.012907228928000298|0.03326689862173776|\n",
      "|min    |463              |472              |40                |0.277               |0.214               |0.301              |\n",
      "|max    |1009             |1103             |116               |0.373               |0.294               |0.491              |\n",
      "+-------+-----------------+-----------------+------------------+--------------------+--------------------+-------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#view statistical measures of data \n",
    "df.describe().show(5,False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[Row(var_1=734, var_2=688, var_3=81, var_4=0.328, var_5=0.259, output=0.418),\n",
       " Row(var_1=700, var_2=600, var_3=94, var_4=0.32, var_5=0.247, output=0.389),\n",
       " Row(var_1=712, var_2=705, var_3=93, var_4=0.311, var_5=0.247, output=0.417)]"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#sneak into the dataset\n",
    "df.head(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "#import corr function from pyspark functions\n",
    "from pyspark.sql.functions import corr"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------------+\n",
      "|corr(var_1, output)|\n",
      "+-------------------+\n",
      "| 0.9187399607627283|\n",
      "+-------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# check for correlation\n",
    "df.select(corr('var_1','output')).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "#import vectorassembler to create dense vectors\n",
    "from pyspark.ml.linalg import Vector\n",
    "from pyspark.ml.feature import VectorAssembler"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['var_1', 'var_2', 'var_3', 'var_4', 'var_5', 'output']"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#select the columns to create input vector\n",
    "df.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "#create the vector assembler \n",
    "vec_assmebler=VectorAssembler(inputCols=['var_1', 'var_2', 'var_3', 'var_4', 'var_5'],outputCol='features')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "#transform the values\n",
    "features_df=vec_assmebler.transform(df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- var_1: integer (nullable = true)\n",
      " |-- var_2: integer (nullable = true)\n",
      " |-- var_3: integer (nullable = true)\n",
      " |-- var_4: double (nullable = true)\n",
      " |-- var_5: double (nullable = true)\n",
      " |-- output: double (nullable = true)\n",
      " |-- features: vector (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#validate the presence of dense vectors \n",
    "features_df.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------------------------+\n",
      "|features                      |\n",
      "+------------------------------+\n",
      "|[734.0,688.0,81.0,0.328,0.259]|\n",
      "|[700.0,600.0,94.0,0.32,0.247] |\n",
      "|[712.0,705.0,93.0,0.311,0.247]|\n",
      "|[734.0,806.0,69.0,0.315,0.26] |\n",
      "|[613.0,759.0,61.0,0.302,0.24] |\n",
      "+------------------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#view the details of dense vector\n",
    "features_df.select('features').show(5,False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "#create data containing input features and output column\n",
    "model_df=features_df.select('features','output')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------------------------+------+\n",
      "|features                      |output|\n",
      "+------------------------------+------+\n",
      "|[734.0,688.0,81.0,0.328,0.259]|0.418 |\n",
      "|[700.0,600.0,94.0,0.32,0.247] |0.389 |\n",
      "|[712.0,705.0,93.0,0.311,0.247]|0.417 |\n",
      "|[734.0,806.0,69.0,0.315,0.26] |0.415 |\n",
      "|[613.0,759.0,61.0,0.302,0.24] |0.378 |\n",
      "+------------------------------+------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "model_df.show(5,False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(1232, 2)\n"
     ]
    }
   ],
   "source": [
    "#size of model df\n",
    "print((model_df.count(), len(model_df.columns)))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Split Data - Train & Test sets\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "#split the data into 70/30 ratio for train test purpose\n",
    "train_df,test_df=model_df.randomSplit([0.7,0.3])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(843, 2)\n"
     ]
    }
   ],
   "source": [
    "print((train_df.count(), len(train_df.columns)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(389, 2)\n"
     ]
    }
   ],
   "source": [
    "print((test_df.count(), len(test_df.columns)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+-------------------+\n",
      "|summary|             output|\n",
      "+-------+-------------------+\n",
      "|  count|                843|\n",
      "|   mean| 0.3981637010676153|\n",
      "| stddev|0.03294758324054597|\n",
      "|    min|              0.311|\n",
      "|    max|              0.491|\n",
      "+-------+-------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "train_df.describe().show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Build Linear Regression Model "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [],
   "source": [
    "#Build Linear Regression model \n",
    "lin_Reg=LinearRegression(labelCol='output')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [],
   "source": [
    "#fit the linear regression model on training data set \n",
    "lr_model=lin_Reg.fit(train_df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.19182762797333336"
      ]
     },
     "execution_count": 25,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "lr_model.intercept"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[0.000340839517716,4.79714814651e-05,0.000188236319406,-0.683084289192,0.520925094105]\n"
     ]
    }
   ],
   "source": [
    "print(lr_model.coefficients)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [],
   "source": [
    "training_predictions=lr_model.evaluate(train_df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.0001479324335891862"
      ]
     },
     "execution_count": 28,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "training_predictions.meanSquaredError"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.863563127042712"
      ]
     },
     "execution_count": 29,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "training_predictions.r2"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [],
   "source": [
    "#make predictions on test data \n",
    "test_results=lr_model.evaluate(test_df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+\n",
      "|           residuals|\n",
      "+--------------------+\n",
      "|-0.01339317627364...|\n",
      "|0.005852942292043195|\n",
      "|-0.01270570081380...|\n",
      "|-0.00719060666608...|\n",
      "|-0.00117340799738...|\n",
      "|0.003662428381860...|\n",
      "|-0.01232056280684...|\n",
      "|-0.00201734580289...|\n",
      "|0.011105011921430763|\n",
      "|-0.00108724755114...|\n",
      "+--------------------+\n",
      "only showing top 10 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#view the residual errors based on predictions \n",
    "test_results.residuals.show(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.8796832901372513"
      ]
     },
     "execution_count": 32,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#coefficient of determination value for model\n",
    "test_results.r2"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.011751649289716636"
      ]
     },
     "execution_count": 33,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "test_results.rootMeanSquaredError"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.0001381012610284975"
      ]
     },
     "execution_count": 34,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "test_results.meanSquaredError"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
